package blockchain.consensus.pbft;

import blockchain.dao.DAOService;
import blockchain.dao.rocksDB.RocksdbDAO;
import blockchain.entity.MerkleTree.MerkleTree;
import blockchain.entity.block.Block;
import blockchain.entity.block.BlockBody;
import blockchain.entity.block.BlockHeader;
import blockchain.entity.block.ContentInfo;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.java_websocket.WebSocket;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.DependsOn;
import org.springframework.stereotype.Component;
import org.testng.util.Strings;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * @Author: 李浩然
 * @Date: 2020/12/15 10:51 上午
 */
@Component
public class P2pPointPbftClient {
    private Logger logger = LoggerFactory.getLogger(P2pPointPbftClient.class);
    //P2P网络中的节点既是客户端，又是服务端，服务端运行在端口，又作为客户端通过该端口连接服务端
    private String wsUrl = "ws://localhost:7001";
    //所有客户端的WebSocket的连接池缓存
    DAOService dao = new RocksdbDAO();
    private List<WebSocket> localSockets = new CopyOnWriteArrayList<>();

    public List<WebSocket> getLocalSockets() {
        return localSockets;
    }

    public void setLocalSockets(List<WebSocket> localSockets) {
        this.localSockets = localSockets;
    }

    /**
     * 连接到服务端
     */
    @Bean
    @DependsOn("initServer")
    public void connectPeer() throws URISyntaxException {
        try {
            //创建WebSocket客户端
            final WebSocketClient socketClient = new WebSocketClient(new URI(wsUrl)) {
                @Override
                public void onOpen(ServerHandshake serverHandshake) {
                    sendMessage(this, "客户端成功创建");
                    localSockets.add(this);
                }

                @Override
                public void onMessage(String msg) {
                    logger.info("客户端收到服务端信息：" + msg);
                    //如果收到不是JSON数据，说明不是PBFT阶段
                    if (!msg.startsWith("{")) {
                        return;
                    }
                    //如果是JSON化数据，说明已经进入PBFT投票阶段
                    JSONObject json = JSON.parseObject(msg);
                    if (!json.containsKey("code")) {
                        logger.info("未收到投票状态码！！！");
                    }
                    int code = json.getIntValue("code");
                    if (code == VoteEnum.PREPREPARE.getCode()) {
                        //校验哈希
                        VoteInfo voteInfo = JSON.parseObject(msg, VoteInfo.class);
                        if (!voteInfo.getHash().equals(new MerkleTree(voteInfo.getList()).getRoot().getHash())) {
                            logger.info("未通过Merkle树Hash校验");
                            return;
                        }
                        //若校验成功，进入下一个状态
                        voteInfo.setCode(VoteEnum.PREPARE.getCode());
                        sendMessage(this, JSON.toJSONString(voteInfo));
                        logger.info("客户端发送到服务端pbft消息：" + JSON.toJSONString(voteInfo));
                        return;
                    }

                    if (code == VoteEnum.COMMIT.getCode()) {
                        //校验哈希
                        VoteInfo voteInfo = JSON.parseObject(msg, VoteInfo.class);
                        if (!voteInfo.getHash().equals(new MerkleTree(voteInfo.getList()).getRoot().getHash())) {
                            logger.info("未通过Merkle树Hash校验");
                            return;
                        }
                        //校验成功,检验节点个数是否有效
                        if (getConnecttedNodeCount() >= getLeastNodeCount()) {
                            sendMessage(this, "客户端开始写入区块");
                            logger.info("客户端开始写入区块");
                        }
                        Block block = new Block();
                        List<String> contentList = new ArrayList<>();
                        for (ContentInfo contentInfo : block.getBlockBody().getContentInfos()) {
                            contentList.add(contentInfo.getJsonContent());
                        }
                        block.setBlockHash(new MerkleTree(contentList).getRoot().getHash());
                        block.setBlockHeader(new BlockHeader());
                        block.setBlockBody(new BlockBody());
                        dao.put(voteInfo.getHash(), voteInfo.toString());
                    }
                }

                @Override
                public void onClose(int i, String s, boolean b) {
                    logger.info("客户端关闭");
                    localSockets.remove(this);
                }

                @Override
                public void onError(Exception e) {
                    logger.info("客户端报错");
                    e.printStackTrace();
                    localSockets.remove(this);
                }
            };
            socketClient.connect();
        } catch (URISyntaxException e) {
            logger.info("连接错误：" + e.getMessage());
        }
    }

    //构建投票信息
    private VoteInfo createVoteInfo(VoteEnum ve, String value) {
        VoteInfo vi = new VoteInfo();
        vi.setCode(ve.getCode());
        List<String> contents = new ArrayList<>();
        contents.add(value);
        String hash = new MerkleTree(contents).getRoot().getHash();
        vi.setList(contents);
        vi.setHash(hash);
        return vi;
    }

    /**
     * 已经连接的节点个数
     *
     * @return
     */
    private int getConnecttedNodeCount() {
        //本机测试为1，多节点部署时按具体情况返回
        return 1;
    }

    /**
     * PBFT消息节点最少确认个数计算
     *
     * @return
     */
    private int getLeastNodeCount() {
        //本机测试时写死为1，实际PBFT拜占庭节点数量f，总节点数量3f+1
        return 1;
    }

    /**
     * 向服务端发送消息，当前WebSocket的远程Socket地址就是服务端
     *
     * @param ws
     * @param message
     */
    public void sendMessage(WebSocket ws, String message) {
        logger.info("发送给：" + ws.getRemoteSocketAddress().getPort() + "的P2P消息：" + message);
        ws.send(message);
    }

    /**
     * 向所有连接过的服务端广播消息
     *
     * @param message:待广播消息
     */
    public void broadcast(String message) {
        if (localSockets.size() == 0 || Strings.isNullOrEmpty(message)) {
            return;
        }
        logger.info("开始广播");
        for (WebSocket socket : localSockets) {
            this.sendMessage(socket, message);
        }
        logger.info("广播结束");
    }

}
